
#include "oc_rocksdb.h"


	KeyValueDb::KeyValueDb() {}

	KeyValueDb::~KeyValueDb() {}

	std::string KeyValueDb::error_desc() {
		return error_desc_;
	}

	RocksDbDriver::RocksDbDriver() {
		db_ = NULL;
	}

	RocksDbDriver::~RocksDbDriver() {
		if (db_ != NULL) {
			delete db_;
			db_ = NULL;
		}
	}

	bool RocksDbDriver::Open(const std::string &db_path) {
		rocksdb::Options options;
		options.create_if_missing = true;
		// Optimize RocksDB. This is the easiest way to get RocksDB to perform well
		long cpus = sysconf(_SC_NPROCESSORS_ONLN);
		options.IncreaseParallelism(cpus);
		//options.OptimizeLevelStyleCompaction();
		rocksdb::Status status = rocksdb::DB::Open(options, db_path, &db_);
		if (!status.ok()) {
			error_desc_ = status.ToString();
		}
		return status.ok();
	}

	bool RocksDbDriver::Close() {
		delete db_;
		db_ = NULL;
		return true;
	}

	bool RocksDbDriver::Get(const std::string &key, std::string &value) {
		assert(db_ != NULL);
		rocksdb::Status status = db_->Get(rocksdb::ReadOptions(), key, &value);
		return status.ok();
	}

	bool RocksDbDriver::Put(const std::string &key, const std::string &value) {
		assert(db_ != NULL);
		rocksdb::WriteOptions opt;
		opt.sync = true;
		rocksdb::Status status = db_->Put(opt, key, value);
		if (!status.ok()) {
			error_desc_ = status.ToString();
		}
		return status.ok();
	}

	bool RocksDbDriver::Delete(const std::string &key) {
		assert(db_ != NULL);
		rocksdb::WriteOptions opt;
		opt.sync = true;
		rocksdb::Status status = db_->Delete(opt, key);
		if (!status.ok()) {
			error_desc_ = status.ToString();
		}
		return status.ok();
	}

	bool RocksDbDriver::WriteBatch(WRITE_BATCH &write_batch) {

		rocksdb::WriteOptions opt;
		opt.sync = true;
		rocksdb::Status status = db_->Write(opt, &write_batch);
		if (!status.ok()) {
			error_desc_ = status.ToString();
		}
		return status.ok();
	}

	void* RocksDbDriver::NewIterator() {
		return db_->NewIterator(rocksdb::ReadOptions());
	}

	bool RocksDbDriver::GetOptions(Json::Value &options) {
		std::string out;
		db_->GetProperty("rocksdb.estimate-table-readers-mem", &out);
		options["rocksdb.estimate-table-readers-mem"] = out;

		db_->GetProperty("rocksdb.cur-size-all-mem-tables", &out);
		options["rocksdb.cur-size-all-mem-tables"] = out;
		return true;
	}

	RationalDb::RationalDb() {}

	RationalDb::~RationalDb() {}

	std::string RationalDb::Format(const std::string &value) {
		std::string new_value = value;
		OCUtils::String::Replace(new_value, "\\", "\\\\");
		OCUtils::String::Replace(new_value, "'", "\\'");
		OCUtils::String::Replace(new_value, "\"", "\\\"");
		OCUtils::String::Replace(new_value, "\r", "\\r");
		OCUtils::String::Replace(new_value, "\n", "\\n");

		return new_value;
	}


	int32_t RationalDb::QueryRecord(const std::string &table, const std::string &sqlwhere, Json::Value &record) {
		std::string sql = OCUtils::String::Format("select * from %s %s", table.c_str(), sqlwhere.c_str());
		return QueryRecord(sql, record);
	}

	Storage::Storage() {
		keyvalue_db_ = NULL;
		rational_db_ = NULL;
	}

	Storage::~Storage() {}

	bool Storage::Initialize(const std::string &keyvalue_db_path, const std::string &rational_db_con, const std::string &rational_db_type, bool bdropdb) {
		do {
			std::string strConnect = "";
			std::string str_dbname = "";
			std::vector<std::string> nparas = OCUtils::String::split(rational_db_con, " ");
			for (std::size_t i = 0; i < nparas.size(); i++) {
				std::string str = nparas[i];
				std::vector<std::string> n = OCUtils::String::split(str, "=");
				if (n.size() == 2) {
					if (n[0] != "dbname") {
						strConnect += " ";
						strConnect += str;
					}
					else {
						str_dbname = n[1];
					}
				}
			}
			if (bdropdb) {
				RationalDb *rational_db = NULL;

				rational_db = new SociDriver();

				bool do_success = false;
				do {
					if (OCUtils::File::IsExist(keyvalue_db_path) && !OCUtils::File::DeleteFolder(keyvalue_db_path)) {
						LOG_ERROR_ERRNO("Delete keyvalue db failed", STD_ERR_CODE, STD_ERR_DESC);
					}
					if (!rational_db->Open(strConnect)) {
						LOG_ERROR_ERRNO("Delete rational db failed", rational_db->error_code(), rational_db->error_desc());
						rational_db->Close();
						delete rational_db;
						return false;
					}

					std::string cmd = OCUtils::String::Format("drop database if exists %s", str_dbname.c_str());
					if (!rational_db->Execute(cmd)) {
						LOG_ERROR_ERRNO("Delete rational db failed", rational_db->error_code(), rational_db->error_desc());
						break;
					}

					//delete tmp files
					std::string path = Configure::Instance().db_configure_.tmp_path_;
					OCUtils::FileAttributes nlist;
					OCUtils::File::GetFileList(path, nlist);
					for (auto it = nlist.begin(); it != nlist.end(); ++it) {
						if (it->first.rfind(bubi::General::RATIONAL_TMPDB_POSTFIX) != std::string::npos
							|| it->first.rfind(bubi::General::KEYVALUE_TMPDB_POSTFIX) != std::string::npos) {
							std::string filename = OCUtils::String::Format("%s/%s", path.c_str(), it->first.c_str());
							LOG_INFO("removing tmp file %s", filename.c_str());
							OCUtils::File::Delete(filename);
						}
					}

					LOG_INFO("Drop db successful");
					do_success = true;
				} while (false);

				rational_db->Close();
				delete rational_db;
				return do_success;
			}


			keyvalue_db_ = new RocksDbDriver();
			if (!keyvalue_db_->Open(keyvalue_db_path)) {
				LOG_ERROR("keyvalue_db path(%s) open fail(%s)\n",
					keyvalue_db_path.c_str(), keyvalue_db_->error_desc().c_str());
				break;
			}
			if (!OCUtils::File::IsExist(Configure::Instance().db_configure_.tmp_path_)) {
				OCUtils::File::CreateDir(Configure::Instance().db_configure_.tmp_path_);
			}
			rational_db_ = NULL;

			rational_db_ = new SociDriver();

			//save the rational db type
			ratiional_db_type_ = rational_db_type;

			if (!rational_db_->Open(rational_db_con)) {
				if (!rational_db_->Open(strConnect)) {
					LOG_ERROR_ERRNO("open rational db failed", rational_db_->error_code(), rational_db_->error_desc());
					break;
				}

				std::string sql = OCUtils::String::Format("create database %s", str_dbname.c_str());
				if (!rational_db_->Execute(sql)) {
					LOG_ERROR_ERRNO("create database failed", rational_db_->error_code(), rational_db_->error_desc());
					break;
				}

				if (!rational_db_->Open(rational_db_con)) {
					LOG_ERROR_ERRNO("open rational db failed", rational_db_->error_code(), rational_db_->error_desc());
					break;
				}
			}
			return true;
		} while (false);

		CloseDb();

		return false;
	}


	bool  Storage::CloseDb() {
		bool ret1 = true, ret2 = true;
		if (keyvalue_db_ != NULL) {
			ret1 = keyvalue_db_->Close();
			delete keyvalue_db_;
			keyvalue_db_ = NULL;
		}

		if (rational_db_ != NULL) {
			ret2 = rational_db_->Close();
			delete rational_db_;
			rational_db_ = NULL;
		}

		return ret1 && ret2;
	}

	bool Storage::Exit() {
		return CloseDb();
	}

	KeyValueDb *Storage::keyvalue_db() {
		return keyvalue_db_;
	}

	RationalDb *Storage::rational_db() {
		return rational_db_;
	}

	std::shared_ptr<RationalDb> Storage::NewRationalDb() {
		DbConfigure  &config = bubi::Configure::Instance().db_configure_;
		RationalDb *db = NULL;

		db = new SociDriver();

		do {
			if (!db->Open(config.soci_connection_)) {
				LOG_ERROR("open rational db failed(%s)", rational_db_->error_desc());
				break;
			}

			return std::shared_ptr<RationalDb>(db);

		} while (false);

		return NULL;
	}

	SociDriver::SociDriver(void) {

	}

	SociDriver::~SociDriver(void) {

	}

	bool SociDriver::CreateTable(const std::string &table_name, const Json::Value &columns) {
		return false;
	}

	std::string SociDriver::GetSqls() {
		return sql_;
	}

	bool SociDriver::Open(const DbConfigure &cfg) {
		std::string str = OCUtils::String::Format("dbname=%s user=%s password=%s",
			cfg.soci_dbname_.c_str(), cfg.soci_user_.c_str(), cfg.soci_password_.c_str());
		session_ = PQconnectdb(str.c_str());

		if (session_ != nullptr) {
			return true;
		}
		else {
			LOG_ERROR("connect error!");
			return false;
		}

	}


	bool SociDriver::Open(const std::string &connect) {
		session_ = PQconnectdb(connect.c_str());
		if (PQstatus(session_) != CONNECTION_OK) {
			error_code_ = PQstatus(session_);
			error_desc_ = PQerrorMessage(session_);
			return false;
		}
		else {
			return true;
		}
	}

	bool SociDriver::Execute(const std::string &sql) {
		OCUtils::MutexGuard guard(mutex_);
		PGresult* res = PQexec(session_, sql.c_str());
		ExecStatusType statustype = PQresultStatus(res);

		switch (statustype) {
		case PGRES_COMMAND_OK:
		case PGRES_TUPLES_OK:
			PQclear(res);
			return true;
		case PGRES_EMPTY_QUERY:
		case PGRES_COPY_OUT:
		case PGRES_COPY_IN:
		case PGRES_BAD_RESPONSE:
		case PGRES_NONFATAL_ERROR:
		case PGRES_FATAL_ERROR:
			PQclear(res);
			error_code_ = statustype;
			error_desc_ = PQerrorMessage(session_);
			return false;
		default:
			PQclear(res);
			error_code_ = statustype;
			error_desc_ = PQerrorMessage(session_);
			return false;
		}
	}

	bool SociDriver::Insert(const std::string &table_name, const OCUtils::StringMap &record) {
		std::string names, values;
		for (OCUtils::StringMap::const_iterator iter = record.begin();
			iter != record.end();
			iter++) {
			if (names.empty()) {
				names = OCUtils::String::Format("%s", Format(iter->first).c_str());
				values = OCUtils::String::Format("'%s'", Format(iter->second).c_str());
			}
			else {
				names = OCUtils::String::Format("%s,%s", names.c_str(), Format(iter->first).c_str());
				values = OCUtils::String::Format("%s,'%s'", values.c_str(), Format(iter->second).c_str());
			}
		}

		std::string sql = OCUtils::String::Format("INSERT INTO %s(%s) VALUES(%s)", table_name.c_str(), names.c_str(), values.c_str());
		return Execute(sql);
	}

	bool SociDriver::Close() {
		PQfinish(session_);
		return true;
	}

	int32_t SociDriver::Query(const std::string &sql, Json::Value &record_array) {
		OCUtils::MutexGuard guard(mutex_);

		int32_t nrows = 0;
		PGresult* res = PQexec(session_, sql.c_str());
		ExecStatusType status = PQresultStatus(res);
		if (status != PGRES_TUPLES_OK) {
			error_code_ = status;
			error_desc_ = PQerrorMessage(session_);
			PQclear(res);
			return -1;
		}
		int row_count = PQntuples(res);
		int col_count = PQnfields(res);

		for (int row = 0; row < row_count; row++) {
			for (int col = 0; col < col_count; col++) {
				std::string colname(PQfname(res, col));
				char * buf = PQgetvalue(res, row, col);
				int length = PQgetlength(res, row, col);
				Oid ctype = PQftype(res, col);
				switch (ctype) {
				case 25:   // text
				case 1043: // varchar
				case 2275: // cstring
				case 18:   // char
				case 1042: // bpchar
				case 142: // xml
				case 114:  // json
				case 17: // bytea
				case 2950: // uuid
				{
					record_array[row][colname] = std::string(buf, length);
					break;
				}

				case 702:  // abstime
				case 703:  // reltime
				case 1082: // date
				case 1083: // time
				case 1114: // timestamp
				case 1184: // timestamptz
				case 1266: // timetz
				{
					record_array[row][colname] = std::string(buf, length);
					break;
				}

				case 700:  // float4
				case 701:  // float8
				case 1700: // numeric
				{
					record_array[row][colname] = OCUtils::String::Stod(std::string(buf, length));
					break;
				}

				case 16:   // bool
				case 21:   // int2
				case 23:   // int4
				case 26:   // oid
				{
					record_array[row][colname] = OCUtils::String::Stoi64(std::string(buf, length));
					break;
				}
				case 20:   // int8
				{
					record_array[row][colname] = OCUtils::String::Stoi64(std::string(buf, length));
					break;
				}
				default:
				{
				}
				}
			}
		}
		PQclear(res);
		return row_count;
	}

	int32_t SociDriver::QueryRecord(const std::string &sql, Json::Value &record) {
		OCUtils::MutexGuard guard(mutex_);

		PGresult* res = PQexec(session_, sql.c_str());
		ExecStatusType status = PQresultStatus(res);
		if (status != PGRES_TUPLES_OK) {
			error_code_ = status;
			error_desc_ = PQerrorMessage(session_);
			PQclear(res);
			return -1;
		}
		int row_count = PQntuples(res);
		int col_count = PQnfields(res);
		if (row_count <= 0) {
			return row_count;
		}

		for (int col = 0; col < col_count; col++) {
			std::string colname(PQfname(res, col));
			char * buf = PQgetvalue(res, 0, col);
			Oid ctype = PQftype(res, col);
			int length = PQgetlength(res, 0, col);
			switch (ctype) {
			case 25:   // text
			case 1043: // varchar
			case 2275: // cstring
			case 18:   // char
			case 1042: // bpchar
			case 142: // xml
			case 114:  // json
			case 17: // bytea
			case 2950: // uuid
			{
				record[colname] = std::string(buf, length);
				break;
			}

			case 702:  // abstime
			case 703:  // reltime
			case 1082: // date
			case 1083: // time
			case 1114: // timestamp
			case 1184: // timestamptz
			case 1266: // timetz
			{
				record[colname] = std::string(buf, length);
				break;
			}

			case 700:  // float4
			case 701:  // float8
			case 1700: // numeric
			{
				std::string s = std::string(buf, length);
				record[colname] = OCUtils::String::Stod(s);
				break;
			}

			case 16:   // bool
			case 21:   // int2
			case 23:   // int4
			case 26:   // oid
			{
				std::string s = std::string(buf, length);
				record[colname] = OCUtils::String::Stoi64(s);
				break;
			}
			case 20:   // int8
			{
				int length = PQgetlength(res, 0, col);
				record[colname] = OCUtils::String::Stoi64(std::string(buf, length));
				break;
			}
			default:
			{
			}
			}
		}

		PQclear(res);
		return row_count;
	}

	bool SociDriver::Update(const std::string &table_name, const OCUtils::StringMap &record, const std::string &sql_where) {
		std::string sets;
		for (OCUtils::StringMap::const_iterator iter = record.begin();
			iter != record.end();
			iter++) {
			if (sets.empty()) {
				sets = OCUtils::String::Format("%s=%s", Format(iter->first).c_str(), Format(iter->second).c_str());
			}
			else {
				sets = OCUtils::String::Format("%s,%s=%s", sets.c_str(), Format(iter->first).c_str(), Format(iter->second).c_str());
			}
		}

		std::string sql = OCUtils::String::Format("UPDATE %s SET %s %s", table_name.c_str(), sets.c_str(), sql_where.c_str());
		return Execute(sql);
	}

	void SociDriver::SetError(const char *pMessage) {
		
	}
	const char *SociDriver::error_desc() {
		return error_desc_.c_str();
	}


	int SociDriver::error_code() {
		return error_code_;
	}

	int32_t SociDriver::DescribeTable(const std::string &table_name, Json::Value &table_desc) {
		std::string strcond = OCUtils::String::Format("where relname='%s'", table_name.c_str());
		int64_t row_count = QueryCount("pg_class", strcond);
		if (row_count <= 0) {
			return 0;
		}
		return 1;
	}

	int64_t SociDriver::QueryCount(const std::string &table_name, const std::string &condition) {
		OCUtils::MutexGuard guard(mutex_);

		std::string sql = OCUtils::String::Format("SELECT COUNT(*) AS count FROM %s %s", table_name.c_str(), condition.c_str());
		int64_t count = 0;
		PGresult* res = PQexec(session_, sql.c_str());
		ExecStatusType ty = PQresultStatus(res);
		if (ty != PGRES_TUPLES_OK) {
			error_code_ = ty;
			error_desc_ = PQerrorMessage(session_);
			PQclear(res);
			return -1;
		}
		char * buf = PQgetvalue(res, 0, 0);
		int len = PQgetlength(res, 0, 0);
		count = OCUtils::String::Stoi64(std::string(buf, len));
		PQclear(res);
		return count;
	}

	void SociDriver::AppendSql(const std::string &sql) {
		sql_ += sql;
		sql_ += ";";
	}

	bool SociDriver::Begin() {
		sql_ = "BEGIN;";
		return true;
	}

	bool SociDriver::Commit() {
		sql_ += "COMMIT;";
		return Execute(sql_);
	}
